/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.quorum;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.jmx.MBeanRegistry;
import org.apache.zookeeper.server.quorum.QuorumCnxManager.Message;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Implementation of leader election using TCP. It uses an object of the class
 * QuorumCnxManager to manage connections. Otherwise, the algorithm is
 * push-based as with the other UDP implementations.
 *
 * There are a few parameters that can be tuned to change its behavior. First,
 * finalizeWait determines the amount of time to wait until deciding upon a
 * leader. This is part of the leader election algorithm.
 */

public class FastLeaderElection implements Election {
	/**
	 * Multi-threaded implementation of message handler. Messenger implements two
	 * sub-classes: WorkReceiver and WorkSender. The functionality of each is
	 * obvious from the name. Each of these spawns a new thread.
	 */

	private class Messenger {

		/**
		 * Receives messages from instance of QuorumCnxManager on method run(), and
		 * processes such messages.
		 */

		class WorkerReceiver implements Runnable {
			QuorumCnxManager manager;
			volatile boolean stop;

			WorkerReceiver(QuorumCnxManager manager) {
				this.stop = false;
				this.manager = manager;
			}

			@Override
			public void run() {

				Message response;
				while (!this.stop) {
					// Sleeps on receive
					try {
						response = this.manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
						if (response == null)
							continue;

						/*
						 * If it is from an observer, respond right away. Note that the following
						 * predicate assumes that if a server is not a follower, then it must be an
						 * observer. If we ever have any other type of learner in the future, we'll have
						 * to change the way we check for observers.
						 */
						if (!FastLeaderElection.this.self.getVotingView().containsKey(response.sid)) {
							Vote current = FastLeaderElection.this.self.getCurrentVote();
							ToSend notmsg = new ToSend(ToSend.mType.notification, current.getId(), current.getZxid(),
									FastLeaderElection.this.logicalclock, FastLeaderElection.this.self.getPeerState(), response.sid, current.getPeerEpoch());

							FastLeaderElection.this.sendqueue.offer(notmsg);
						} else {
							// Receive new message
							if (LOG.isDebugEnabled()) {
								LOG.debug("Receive new notification message. My id = " + FastLeaderElection.this.self.getId());
							}

							/*
							 * We check for 28 bytes for backward compatibility
							 */
							if (response.buffer.capacity() < 28) {
								LOG.error("Got a short response: " + response.buffer.capacity());
								continue;
							}
							boolean backCompatibility = (response.buffer.capacity() == 28);
							response.buffer.clear();

							// State of peer that sent this message
							QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
							switch (response.buffer.getInt()) {
							case 0:
								ackstate = QuorumPeer.ServerState.LOOKING;
								break;
							case 1:
								ackstate = QuorumPeer.ServerState.FOLLOWING;
								break;
							case 2:
								ackstate = QuorumPeer.ServerState.LEADING;
								break;
							case 3:
								ackstate = QuorumPeer.ServerState.OBSERVING;
								break;
							}

							// Instantiate Notification and set its attributes
							Notification n = new Notification();
							n.leader = response.buffer.getLong();
							n.zxid = response.buffer.getLong();
							n.electionEpoch = response.buffer.getLong();
							n.state = ackstate;
							n.sid = response.sid;
							if (!backCompatibility) {
								n.peerEpoch = response.buffer.getLong();
							} else {
								if (LOG.isInfoEnabled()) {
									LOG.info("Backward compatibility mode, server id=" + n.sid);
								}
								n.peerEpoch = ZxidUtils.getEpochFromZxid(n.zxid);
							}

							/*
							 * Print notification info
							 */
							if (LOG.isInfoEnabled()) {
								FastLeaderElection.this.printNotification(n);
							}

							/*
							 * If this server is looking, then send proposed leader
							 */

							if (FastLeaderElection.this.self.getPeerState() == QuorumPeer.ServerState.LOOKING) {
								FastLeaderElection.this.recvqueue.offer(n);

								/*
								 * Send a notification back if the peer that sent this message is also looking
								 * and its logical clock is lagging behind.
								 */
								if ((ackstate == QuorumPeer.ServerState.LOOKING) && (n.electionEpoch < FastLeaderElection.this.logicalclock)) {
									Vote v = FastLeaderElection.this.getVote();
									ToSend notmsg = new ToSend(ToSend.mType.notification, v.getId(), v.getZxid(),
											FastLeaderElection.this.logicalclock, FastLeaderElection.this.self.getPeerState(), response.sid, v.getPeerEpoch());
									FastLeaderElection.this.sendqueue.offer(notmsg);
								}
							} else {
								/*
								 * If this server is not looking, but the one that sent the ack is looking, then
								 * send back what it believes to be the leader.
								 */
								Vote current = FastLeaderElection.this.self.getCurrentVote();
								if (ackstate == QuorumPeer.ServerState.LOOKING) {
									if (LOG.isDebugEnabled()) {
										LOG.debug("Sending new notification. My id =  " + FastLeaderElection.this.self.getId() + " recipient="
												+ response.sid + " zxid=0x" + Long.toHexString(current.getZxid())
												+ " leader=" + current.getId());
									}
									ToSend notmsg = new ToSend(ToSend.mType.notification, current.getId(),
											current.getZxid(), FastLeaderElection.this.logicalclock, FastLeaderElection.this.self.getPeerState(), response.sid,
											current.getPeerEpoch());
									FastLeaderElection.this.sendqueue.offer(notmsg);
								}
							}
						}
					} catch (InterruptedException e) {
						System.out.println("Interrupted Exception while waiting for new message" + e.toString());
					}
				}
				LOG.info("WorkerReceiver is down");
			}
		}

		/**
		 * This worker simply dequeues a message to send and and queues it on the
		 * manager's queue.
		 */

		class WorkerSender implements Runnable {
			QuorumCnxManager manager;
			volatile boolean stop;

			WorkerSender(QuorumCnxManager manager) {
				this.stop = false;
				this.manager = manager;
			}

			/**
			 * Called by run() once there is a new message to send.
			 *
			 * @param m message to send
			 */
			private void process(ToSend m) {
				byte requestBytes[] = new byte[36];
				ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);

				/*
				 * Building notification packet to send
				 */

				requestBuffer.clear();
				requestBuffer.putInt(m.state.ordinal());
				requestBuffer.putLong(m.leader);
				requestBuffer.putLong(m.zxid);
				requestBuffer.putLong(m.electionEpoch);
				requestBuffer.putLong(m.peerEpoch);

				this.manager.toSend(m.sid, requestBuffer);

			}

			@Override
			public void run() {
				while (!this.stop) {
					try {
						ToSend m = FastLeaderElection.this.sendqueue.poll(3000, TimeUnit.MILLISECONDS);
						if (m == null)
							continue;

						this.process(m);
					} catch (InterruptedException e) {
						break;
					}
				}
				LOG.info("WorkerSender is down");
			}
		}

		WorkerReceiver wr;

		WorkerSender ws;

		/**
		 * Constructor of class Messenger.
		 *
		 * @param manager Connection manager
		 */
		Messenger(QuorumCnxManager manager) {

			this.ws = new WorkerSender(manager);

			Thread t = new Thread(this.ws, "WorkerSender[myid=" + FastLeaderElection.this.self.getId() + "]");
			t.setDaemon(true);
			t.start();

			this.wr = new WorkerReceiver(manager);

			t = new Thread(this.wr, "WorkerReceiver[myid=" + FastLeaderElection.this.self.getId() + "]");
			t.setDaemon(true);
			t.start();
		}

		/**
		 * Stops instances of WorkerSender and WorkerReceiver
		 */
		void halt() {
			this.ws.stop = true;
			this.wr.stop = true;
		}

		/**
		 * Test if both send and receive queues are empty.
		 */
		public boolean queueEmpty() {
			return (FastLeaderElection.this.sendqueue.isEmpty() || FastLeaderElection.this.recvqueue.isEmpty());
		}

	}

	/**
	 * Notifications are messages that let other peers know that a given peer has
	 * changed its vote, either because it has joined leader election or because it
	 * learned of another peer with higher zxid or same zxid and higher server id
	 */

	static public class Notification {
		/*
		 * Epoch
		 */
		long electionEpoch;

		/*
		 * Proposed leader
		 */
		long leader;

		/*
		 * epoch of the proposed leader
		 */
		long peerEpoch;

		/*
		 * Address of sender
		 */
		long sid;

		/*
		 * current state of sender
		 */
		QuorumPeer.ServerState state;

		/*
		 * zxid of the proposed leader
		 */
		long zxid;
	}

	/**
	 * Messages that a peer wants to send to other peers. These messages can be both
	 * Notifications and Acks of reception of notification.
	 */
	static public class ToSend {
		static enum mType {
			ack, challenge, crequest, notification
		}

		/*
		 * Epoch
		 */
		long electionEpoch;

		/*
		 * Proposed leader in the case of notification
		 */
		long leader;

		/*
		 * Leader epoch
		 */
		long peerEpoch;

		/*
		 * Address of recipient
		 */
		long sid;

		/*
		 * Current state;
		 */
		QuorumPeer.ServerState state;

		/*
		 * id contains the tag for acks, and zxid for notifications
		 */
		long zxid;

		ToSend(mType type, long leader, long zxid, long electionEpoch, ServerState state, long sid, long peerEpoch) {

			this.leader = leader;
			this.zxid = zxid;
			this.electionEpoch = electionEpoch;
			this.state = state;
			this.sid = sid;
			this.peerEpoch = peerEpoch;
		}
	}

	/**
	 * Determine how much time a process has to wait once it believes that it has
	 * reached the end of leader election.
	 */
	final static int finalizeWait = 200;

	private static final Logger LOG = LoggerFactory.getLogger(FastLeaderElection.class);

	/**
	 * Upper bound on the amount of time between two consecutive notification
	 * checks. This impacts the amount of time to get the system up again after long
	 * partitions. Currently 60 seconds.
	 */

	final static int maxNotificationInterval = 60000;

	volatile long logicalclock; /* Election instance */
	/**
	 * Connection manager. Fast leader election uses TCP for communication between
	 * peers, and QuorumCnxManager manages such connections.
	 */

	QuorumCnxManager manager;

	Messenger messenger;

	long proposedEpoch;
	long proposedLeader;
	long proposedZxid;
	LinkedBlockingQueue<Notification> recvqueue;
	QuorumPeer self;
	LinkedBlockingQueue<ToSend> sendqueue;

	volatile boolean stop;

	/**
	 * Constructor of FastLeaderElection. It takes two parameters, one is the
	 * QuorumPeer object that instantiated this object, and the other is the
	 * connection manager. Such an object should be created only once by each peer
	 * during an instance of the ZooKeeper service.
	 *
	 * @param self    QuorumPeer that created this object
	 * @param manager Connection manager
	 */
	public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager) {
		this.stop = false;
		this.manager = manager;
		this.starter(self, manager);
	}

	/**
	 * In the case there is a leader elected, and a quorum supporting this leader,
	 * we have to check if the leader has voted and acked that it is leading. We
	 * need this check to avoid that peers keep electing over and over a peer that
	 * has crashed and it is no longer leading.
	 *
	 * @param votes         set of votes
	 * @param leader        leader id
	 * @param electionEpoch epoch id
	 */
	private boolean checkLeader(HashMap<Long, Vote> votes, long leader, long electionEpoch) {

		boolean predicate = true;

		/*
		 * If everyone else thinks I'm the leader, I must be the leader. The other two
		 * checks are just for the case in which I'm not the leader. If I'm not the
		 * leader and I haven't received a message from leader stating that it is
		 * leading, then predicate is false.
		 */

		if (leader != this.self.getId()) {
			if (votes.get(leader) == null)
				predicate = false;
			else if (votes.get(leader).getState() != ServerState.LEADING)
				predicate = false;
		}

		return predicate;
	}

	public QuorumCnxManager getCnxManager() {
		return this.manager;
	}

	/**
	 * Returns the initial vote value of server identifier.
	 *
	 * @return long
	 */
	private long getInitId() {
		if (this.self.getLearnerType() == LearnerType.PARTICIPANT)
			return this.self.getId();
		else
			return Long.MIN_VALUE;
	}

	/**
	 * Returns initial last logged zxid.
	 *
	 * @return long
	 */
	private long getInitLastLoggedZxid() {
		if (this.self.getLearnerType() == LearnerType.PARTICIPANT)
			return this.self.getLastLoggedZxid();
		else
			return Long.MIN_VALUE;
	}

	/**
	 * Returns the current vlue of the logical clock counter
	 */
	public long getLogicalClock() {
		return this.logicalclock;
	}

	/**
	 * Returns the initial vote value of the peer epoch.
	 *
	 * @return long
	 */
	private long getPeerEpoch() {
		if (this.self.getLearnerType() == LearnerType.PARTICIPANT)
			try {
				return this.self.getCurrentEpoch();
			} catch (IOException e) {
				RuntimeException re = new RuntimeException(e.getMessage());
				re.setStackTrace(e.getStackTrace());
				throw re;
			}
		else
			return Long.MIN_VALUE;
	}

	synchronized Vote getVote() {
		return new Vote(this.proposedLeader, this.proposedZxid, this.proposedEpoch);
	}

	/**
	 * A learning state can be either FOLLOWING or OBSERVING. This method simply
	 * decides which one depending on the role of the server.
	 *
	 * @return ServerState
	 */
	private ServerState learningState() {
		if (this.self.getLearnerType() == LearnerType.PARTICIPANT) {
			LOG.debug("I'm a participant: " + this.self.getId());
			return ServerState.FOLLOWING;
		} else {
			LOG.debug("I'm an observer: " + this.self.getId());
			return ServerState.OBSERVING;
		}
	}

	private void leaveInstance(Vote v) {
		if (LOG.isDebugEnabled()) {
			LOG.debug("About to leave FLE instance: leader=" + v.getId() + ", zxid=0x" + Long.toHexString(v.getZxid())
					+ ", my id=" + this.self.getId() + ", my state=" + this.self.getPeerState());
		}
		this.recvqueue.clear();
	}

	/**
	 * Starts a new round of leader election. Whenever our QuorumPeer changes its
	 * state to LOOKING, this method is invoked, and it sends notifications to all
	 * other peers.
	 */
	@Override
	public Vote lookForLeader() throws InterruptedException {
		try {
			this.self.jmxLeaderElectionBean = new LeaderElectionBean();
			MBeanRegistry.getInstance().register(this.self.jmxLeaderElectionBean, this.self.jmxLocalPeerBean);
		} catch (Exception e) {
			LOG.warn("Failed to register with JMX", e);
			this.self.jmxLeaderElectionBean = null;
		}
		if (this.self.start_fle == 0) {
			this.self.start_fle = System.currentTimeMillis();
		}
		try {
			HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();

			HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();

			int notTimeout = finalizeWait;

			synchronized (this) {
				this.logicalclock++;
				this.updateProposal(this.getInitId(), this.getInitLastLoggedZxid(), this.getPeerEpoch());
			}

			LOG.info("New election. My id =  " + this.self.getId() + ", proposed zxid=0x" + Long.toHexString(this.proposedZxid));
			this.sendNotifications();

			/*
			 * Loop in which we exchange notifications until we find a leader
			 */

			while ((this.self.getPeerState() == ServerState.LOOKING) && (!this.stop)) {
				/*
				 * Remove next notification from queue, times out after 2 times the termination
				 * time
				 */
				Notification n = this.recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);

				/*
				 * Sends more notifications if haven't received enough. Otherwise processes new
				 * notification.
				 */
				if (n == null) {
					if (this.manager.haveDelivered()) {
						this.sendNotifications();
					} else {
						this.manager.connectAll();
					}

					/*
					 * Exponential backoff
					 */
					int tmpTimeOut = notTimeout * 2;
					notTimeout = (tmpTimeOut < maxNotificationInterval ? tmpTimeOut : maxNotificationInterval);
					LOG.info("Notification time out: " + notTimeout);
				} else if (this.self.getVotingView().containsKey(n.sid)) {
					/*
					 * Only proceed if the vote comes from a replica in the voting view.
					 */
					switch (n.state) {
					case LOOKING:
						// If notification > current, replace and send messages out
						if (n.electionEpoch > this.logicalclock) {
							this.logicalclock = n.electionEpoch;
							recvset.clear();
							if (this.totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, this.getInitId(), this.getInitLastLoggedZxid(),
									this.getPeerEpoch())) {
								this.updateProposal(n.leader, n.zxid, n.peerEpoch);
							} else {
								this.updateProposal(this.getInitId(), this.getInitLastLoggedZxid(), this.getPeerEpoch());
							}
							this.sendNotifications();
						} else if (n.electionEpoch < this.logicalclock) {
							if (LOG.isDebugEnabled()) {
								LOG.debug(
										"Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
												+ Long.toHexString(n.electionEpoch) + ", logicalclock=0x"
												+ Long.toHexString(this.logicalclock));
							}
							break;
						} else if (this.totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, this.proposedLeader, this.proposedZxid,
								this.proposedEpoch)) {
							this.updateProposal(n.leader, n.zxid, n.peerEpoch);
							this.sendNotifications();
						}

						if (LOG.isDebugEnabled()) {
							LOG.debug("Adding vote: from=" + n.sid + ", proposed leader=" + n.leader
									+ ", proposed zxid=0x" + Long.toHexString(n.zxid) + ", proposed election epoch=0x"
									+ Long.toHexString(n.electionEpoch));
						}

						recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

						if (this.termPredicate(recvset,
								new Vote(this.proposedLeader, this.proposedZxid, this.logicalclock, this.proposedEpoch))) {

							// Verify if there is any change in the proposed leader
							while ((n = this.recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
								if (this.totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, this.proposedLeader, this.proposedZxid,
										this.proposedEpoch)) {
									this.recvqueue.put(n);
									break;
								}
							}

							/*
							 * This predicate is true once we don't read any new relevant message from the
							 * reception queue
							 */
							if (n == null) {
								this.self.setPeerState(
										(this.proposedLeader == this.self.getId()) ? ServerState.LEADING : this.learningState());

								Vote endVote = new Vote(this.proposedLeader, this.proposedZxid, this.proposedEpoch);
								this.leaveInstance(endVote);
								return endVote;
							}
						}
						break;
					case OBSERVING:
						LOG.debug("Notification from observer: " + n.sid);
						break;
					case FOLLOWING:
					case LEADING:
						/*
						 * Consider all notifications from the same epoch together.
						 */
						if (n.electionEpoch == this.logicalclock) {
							recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
							if (this.termPredicate(recvset,
									new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state))
									&& this.checkLeader(outofelection, n.leader, n.electionEpoch)) {
								this.self.setPeerState((n.leader == this.self.getId()) ? ServerState.LEADING : this.learningState());

								Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
								this.leaveInstance(endVote);
								return endVote;
							}
						}

						/**
						 * Before joining an established ensemble, verify that a majority are following
						 * the same leader.
						 */
						outofelection.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
						if (this.termPredicate(outofelection,
								new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state))
								&& this.checkLeader(outofelection, n.leader, n.electionEpoch)) {
							synchronized (this) {
								this.logicalclock = n.electionEpoch;
								this.self.setPeerState((n.leader == this.self.getId()) ? ServerState.LEADING : this.learningState());
							}
							Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
							this.leaveInstance(endVote);
							return endVote;
						}
						break;
					default:
						LOG.warn("Notification state unrecoginized: " + n.state + " (n.state), " + n.sid + " (n.sid)");
						break;
					}
				} else {
					LOG.warn("Ignoring notification from non-cluster member " + n.sid);
				}
			}
			return null;
		} finally {
			try {
				if (this.self.jmxLeaderElectionBean != null) {
					MBeanRegistry.getInstance().unregister(this.self.jmxLeaderElectionBean);
				}
			} catch (Exception e) {
				LOG.warn("Failed to unregister with JMX", e);
			}
			this.self.jmxLeaderElectionBean = null;
		}
	}

	private void printNotification(Notification n) {
		LOG.info("Notification: " + n.leader + " (n.leader), 0x" + Long.toHexString(n.zxid) + " (n.zxid), 0x"
				+ Long.toHexString(n.electionEpoch) + " (n.round), " + n.state + " (n.state), " + n.sid + " (n.sid), 0x"
				+ Long.toHexString(n.peerEpoch) + " (n.peerEPoch), " + this.self.getPeerState() + " (my state)");
	}

	/**
	 * Send notifications to all peers upon a change in our vote
	 */
	private void sendNotifications() {
		for (QuorumServer server : this.self.getVotingView().values()) {
			long sid = server.id;

			ToSend notmsg = new ToSend(ToSend.mType.notification, this.proposedLeader, this.proposedZxid, this.logicalclock,
					QuorumPeer.ServerState.LOOKING, sid, this.proposedEpoch);
			if (LOG.isDebugEnabled()) {
				LOG.debug("Sending Notification: " + this.proposedLeader + " (n.leader), 0x" + Long.toHexString(this.proposedZxid)
						+ " (n.zxid), 0x" + Long.toHexString(this.logicalclock) + " (n.round), " + sid + " (recipient), "
						+ this.self.getId() + " (myid), 0x" + Long.toHexString(this.proposedEpoch) + " (n.peerEpoch)");
			}
			this.sendqueue.offer(notmsg);
		}
	}

	@Override
	public void shutdown() {
		this.stop = true;
		LOG.debug("Shutting down connection manager");
		this.manager.halt();
		LOG.debug("Shutting down messenger");
		this.messenger.halt();
		LOG.debug("FLE is down");
	}

	/**
	 * This method is invoked by the constructor. Because it is a part of the
	 * starting procedure of the object that must be on any constructor of this
	 * class, it is probably best to keep as a separate method. As we have a single
	 * constructor currently, it is not strictly necessary to have it separate.
	 *
	 * @param self    QuorumPeer that created this object
	 * @param manager Connection manager
	 */
	private void starter(QuorumPeer self, QuorumCnxManager manager) {
		this.self = self;
		this.proposedLeader = -1;
		this.proposedZxid = -1;

		this.sendqueue = new LinkedBlockingQueue<ToSend>();
		this.recvqueue = new LinkedBlockingQueue<Notification>();
		this.messenger = new Messenger(manager);
	}

	/**
	 * Termination predicate. Given a set of votes, determines if have sufficient to
	 * declare the end of the election round.
	 *
	 * @param votes Set of votes
	 * @param l     Identifier of the vote received last
	 * @param zxid  zxid of the the vote received last
	 */
	private boolean termPredicate(HashMap<Long, Vote> votes, Vote vote) {

		HashSet<Long> set = new HashSet<Long>();

		/*
		 * First make the views consistent. Sometimes peers will have different zxids
		 * for a server depending on timing.
		 */
		for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
			if (vote.equals(entry.getValue())) {
				set.add(entry.getKey());
			}
		}

		return this.self.getQuorumVerifier().containsQuorum(set);
	}

	/**
	 * Check if a pair (server id, zxid) succeeds our current vote.
	 * @param id   Server identifier
	 * @param zxid Last zxid observed by the issuer of this vote
	 */
	private boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid,
			long curEpoch) {
		LOG.debug("id: " + newId + ", proposed id: " + curId + ", zxid: 0x" + Long.toHexString(newZxid)
				+ ", proposed zxid: 0x" + Long.toHexString(curZxid));
		if (this.self.getQuorumVerifier().getWeight(newId) == 0) {
			return false;
		}

		return ((newEpoch > curEpoch) || ((newEpoch == curEpoch) && (newZxid > curZxid))
				|| ((newZxid == curZxid) && (newId > curId)));
	}

	synchronized void updateProposal(long leader, long zxid, long epoch) {
		if (LOG.isDebugEnabled()) {
			LOG.debug("Updating proposal: " + leader + " (newleader), 0x" + Long.toHexString(zxid) + " (newzxid), "
					+ this.proposedLeader + " (oldleader), 0x" + Long.toHexString(this.proposedZxid) + " (oldzxid)");
		}
		this.proposedLeader = leader;
		this.proposedZxid = zxid;
		this.proposedEpoch = epoch;
	}
}
